package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.common.utils.CollectionUtils;
import com.alibaba.nacos.common.utils.StringUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.common.redis.CacheService;
import com.heima.model.common.constants.ScheduleConstants;
import com.heima.model.common.dtos.common.ResponseResult;
import com.heima.model.common.dtos.common.TaskDto;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;

@Service
@Transactional
public class TaskServiceImpl implements TaskService {
    @Autowired
    private TaskinfoMapper taskinfoMapper;
    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;
    @Autowired
    private CacheService cacheService;

    /**
     * 添加任务
     * @param taskDto
     * @return
     */
    @Override
    public ResponseResult addTask(TaskDto taskDto) {
        //添加任务到数据库
        Taskinfo taskinfo = new Taskinfo();
        //属性拷贝
        BeanUtils.copyProperties(taskDto, taskinfo);
        taskinfo.setPriority(taskDto.getPriority());
        taskinfo.setExecuteTime(new Date(taskDto.getExecuteTime()));
        int insertResult = taskinfoMapper.insert(taskinfo);
        //设置taskId
        taskDto.setTaskId(taskinfo.getTaskId());
        //添加任务日志到数据库中
        if (insertResult==0) {
            return ResponseResult.errorResult(AppHttpCodeEnum.valueOf("任务未添加到数据库中"));
        }
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskDto, taskinfoLogs);
        taskinfoLogs.setExecuteTime(new Date(taskDto.getExecuteTime()));
        taskinfoLogs.setVersion(1);
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
        insertResult = taskinfoLogsMapper.insert(taskinfoLogs);

        //获取当前系统时间
        Calendar calendar = Calendar.getInstance();
        long currentTime = calendar.getTimeInMillis();
        //获取当前系统时间5分钟之后的时间  毫秒值
        calendar.add(Calendar.MINUTE, 5);
        long next5ScheduleTime = calendar.getTimeInMillis();

        //判断任务日志表是否添加成功，若成功，将任务分时段添加到redis中
        //设置key为topic/future_任务类型_任务优先级
        String key=taskDto.getTaskType()+"_"+taskDto.getPriority();
        if (insertResult>0) {
            if(taskinfo.getExecuteTime().getTime()<=currentTime){
                //如果任务的执行时间小于等于当前时间，存入list
                cacheService.lLeftPush(ScheduleConstants.TOPIC+key, JSON.toJSONString(taskDto));
            }else if(taskinfoLogs.getExecuteTime().getTime()<=next5ScheduleTime){
                //如果任务的执行时间大于当前时间 && 小于等于预设时间（未来5分钟） 存入zset中
                cacheService.zAdd(ScheduleConstants.FUTURE+key,JSON.toJSONString(taskDto),taskDto.getExecuteTime());
            }
        }
        return ResponseResult.okResult(taskDto.getTaskId());
    }

    /**
     * 取消任务
     *
     * @param taskId 任务id
     * @return 取消结果
     */
    @Override
    public ResponseResult cancelTask(long taskId) {
        boolean flag=false;
        //删除数据库中的需要删除的任务
        int deleteResult = taskinfoMapper.deleteById(taskId);
        //如果删除失败则直接返回
        if (deleteResult==0) {
            return ResponseResult.okResult(flag);
        }
        //更改任务日志表中的状态为2(CANCELLED)
        LambdaQueryWrapper<TaskinfoLogs> wrapper=new LambdaQueryWrapper<>();
        wrapper.eq(TaskinfoLogs::getTaskId,taskId);
        TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(wrapper);
        taskinfoLogs.setStatus(ScheduleConstants.CANCELLED);
        int updateResult = taskinfoLogsMapper.updateById(taskinfoLogs);
        //该对象用于删除redis数据时使用
        TaskDto taskDto=new TaskDto();
        BeanUtils.copyProperties(taskinfoLogs,taskDto);
        taskDto.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        //获取当前系统时间
        Calendar calendar = Calendar.getInstance();
        long currentTime = calendar.getTimeInMillis();
        //获取当前时间后五分钟的时间
        calendar.add(Calendar.MINUTE, 5);
        long next5ScheduleTime = calendar.getTimeInMillis();
        //设置key为topic/future_任务类型_任务优先级
        String key=taskinfoLogs.getTaskType()+"_"+taskinfoLogs.getPriority();
        //如果更新成功，则删除redis中的数据
        if (updateResult>0) {
            if(taskinfoLogs.getExecuteTime().getTime()<=currentTime){
                cacheService.lRemove(ScheduleConstants.TOPIC+key,0,JSON.toJSONString(taskDto));
            }else if(taskinfoLogs.getExecuteTime().getTime()<=next5ScheduleTime){
                cacheService.zRemove(ScheduleConstants.FUTURE+key,JSON.toJSONString(taskDto));
            }
            flag=true;
        }
        return ResponseResult.okResult(flag);
    }

    /**
     * 按照类型和优先级来拉取任务
     * @param type
     * @param priority
     * @return
     */
    @Override
    public ResponseResult pull(int type, int priority) {
            TaskDto taskDto=new TaskDto();
            //设置key为topic/future_任务类型_任务优先级
            String key=type+"_"+priority;
            //从list中拉取任务,并将该任务信息放入变量taskDto_json中,拉取后redis中自动删除该任务数据
            String taskDto_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
            //如果拉取成功,则进行如下操作
            if (taskDto_json!=null) {
                //删除任务表中被拉取的任务
                taskDto = JSON.parseObject(taskDto_json, TaskDto.class);
                taskinfoMapper.deleteById(taskDto.getTaskId());
                //更新任务日志表中的状态
                TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskDto.getTaskId());
                taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);
                taskinfoLogsMapper.updateById(taskinfoLogs);
            }else{
                return ResponseResult.okResult(null);
            }
        return ResponseResult.okResult(taskDto);
    }

    /**
     * 未来数据定时刷新到list队列中供消费者读取
     * @Scheduled:cron:设置调度任务，定时每分钟运行一次
     */
    @Scheduled(cron = "0 0/1 * * * ?")
    public void refresh() {
        //设置分布式锁,一方获取锁后，另一方只能等待，只有获锁方执行完毕释放锁，另一方才能获取锁
        String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
        if(StringUtils.isNotBlank(token)){
            //System.out.println(System.currentTimeMillis()/1000+"执行了调度任务");
            //获取所有未来数据的集合的Key
            Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_*
            for (String futureKey : futureKeys) { // EG:"future_250_250" ---> "future_"+"250_250"
                //将满足条件的未来数据队列放入消费者队列中供消费者读取,将key分隔为两部分,见上EG
                String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];
                //获取该组key下当前需要消费的任务数据
                Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
                if (!tasks.isEmpty()) {
                    //将这些任务数据添加到消费者队列中
                    cacheService.refreshWithPipeline(futureKey,topicKey,tasks);
                    System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
                }
            }
        }
    }

    /**
     * 数据库任务定时同步到redis
     * Scheduled:cron:设置调度任务，定时每五钟运行一次
     * PostConstruct:初始化方法,开启启动器就运行这个方法
     */
    //@PostConstruct
    @Scheduled(cron = "0 0/1 * * * ? ")
    public void reloadData() {
        //清理缓存中的数据,防止任务表系统时间+预计时间后的数据与redis中原有数据重复
        Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");
        Set<String> topicKeys = cacheService.scan(ScheduleConstants.TOPIC + "*");
        cacheService.delete(futureKeys);
        cacheService.delete(topicKeys);
        //查询符合条件的任务 查找任务表中(执行时间<未来五分钟【系统时间+预计时间+...的时间】)的数据
        //等同于这些任务已经进入预计时间范围内可以被放入为未来队列中了
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        LambdaQueryWrapper<Taskinfo> wrapper=new LambdaQueryWrapper<>();
        wrapper.lt(Taskinfo::getExecuteTime,calendar.getTime());
        List<Taskinfo> taskInfos = taskinfoMapper.selectList(wrapper);
        //把任务添加到redis中
        if(CollectionUtils.isNotEmpty(taskInfos)){
            for (Taskinfo taskInfo : taskInfos) {
                cancelTask(taskInfo.getTaskId());
                TaskDto taskDto=new TaskDto();
                BeanUtils.copyProperties(taskInfo,taskDto);
                taskDto.setExecuteTime(taskInfo.getExecuteTime().getTime());
                addTask(taskDto);
            }
        }
    }
}
